package day07;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * @Description: mysql
 * @Author: ZYX
 * @Date: 2022/2/18 11:18
 * @Version: 1.0
 */
public class Demo04 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env.fromElements(Tuple2.of("key",1),Tuple2.of("key",2))
                .addSink(new RichSinkFunction<Tuple2<String, Integer>>() {

                    private Connection conn;
                    private PreparedStatement insertState;
                    private PreparedStatement updateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/mysql_test?&useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=Asia/Shanghai",
                                "root",
                                "123456")   ;

                        insertState = conn.prepareStatement("INSERT INTO flink_test(f_key,f_value) VALUES (?,?)");
                        updateState = conn.prepareStatement("UPDATE flink_test SET f_value = ? WHERE f_key = ?");
                    }

                    @Override
                    public void close() throws Exception {
                        super.close();
                        insertState.close();
                        updateState.close();
                        conn.close();
                    }

                    @Override
                    public void invoke(Tuple2<String, Integer> value, Context context) throws Exception {
                        super.invoke(value,context);
                        updateState.setInt(1,value.f1);
                        updateState.setString(2,value.f0);

                        updateState.execute();

                        if (updateState.getUpdateCount() == 0) {
                            insertState.setString(1,value.f0);
                            insertState.setInt(2,value.f1);
                            insertState.execute();
                        }
                    }
                });

        env.execute();
    }

}
